package blue.mqtt4j.internal.client.task;

import blue.mqtt4j.core.handler.SenderListener;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * @author Jin Zheng
 * @since 1.0 2020-10-19
 */
public class ScheduledTaskExecutor
{
	private static Logger logger = LoggerFactory.getLogger(ScheduledTaskExecutor.class);
	private static final int DELAY = 0;
	private static final int PERIOD = 1000;

	private final ConcurrentMap<Integer, ScheduledFuture<?>> futureMap = new ConcurrentHashMap<>();
	private Channel channel;

	public ScheduledTaskExecutor()
	{
	}

	public void schedule(int messageId, MqttMessage message)
	{
		if (channel == null)
		{
			logger.warn("Channel is null");
			return;
		}
		ScheduledFuture<?> future = channel.eventLoop().scheduleAtFixedRate(() ->
		{
			channel.writeAndFlush(message).addListener(new SenderListener(message));
		}, DELAY, PERIOD, TimeUnit.MILLISECONDS);
		futureMap.put(messageId, future);
		logger.debug("Add ScheduledFuture, messageId: {}", messageId);
	}

	public void cancel(int messageId)
	{
		ScheduledFuture<?> future = futureMap.get(messageId);
		if (future == null)
		{
			logger.warn("ScheduledFuture is null, messageId: {}", messageId);
			return;
		}
		if (!future.isCancelled())
		{
			future.cancel(false);
		}
		futureMap.remove(messageId);
		logger.debug("Cancel and remove ScheduledFuture, messageId: {}", messageId);
	}

	public void setChannel(Channel channel)
	{
		this.channel = channel;
	}

}
